package rpc

import (
	"cluster-cache/common"
	"encoding/binary"
	"github.com/panjf2000/gnet/v2/pkg/logging"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

// Conn 发送数据到其他节点,一个节点创建一个Conn, 一个Conn里面包含多个net.Conn
type Conn struct {
	sync.Mutex
	connections []net.Conn
	round       uint32
	maxConn     uint32
	codecs      []*HeadBodyCodec
	target      string
	lastIndex   *common.LastIndex
	start       bool
}

var ConnectTimeout = time.Second

func NewConn(target string, maxConn int, index *common.LastIndex) *Conn {
	conn := &Conn{
		connections: make([]net.Conn, maxConn, maxConn),
		round:       0,
		maxConn:     uint32(maxConn),
		codecs:      make([]*HeadBodyCodec, maxConn, maxConn),
		target:      target,
		lastIndex:   index,
		start:       false,
	}
	for i := uint32(0); i < uint32(maxConn); i++ {
		conn.codecs[i] = NewHeadBodyCodec()
		conn.Reconnect(i)
		go conn.decoding(i)
	}
	conn.start = true
	return conn
}

func (c *Conn) decoding(u uint32) {
	// 设置了lastIndex则说明开启了全双工模式
	if c.lastIndex == nil {
		return
	}
	codec := c.codecs[u]
	conn := c.connections[u]
	for true {
		data, seq, err := codec.Decode(conn)
		if err != nil {
			if err == ErrIncompletePacket {
				continue
			}
			logging.Infof("reconnecting %s because %v", c.target, err)
			conn = c.Reconnect(u)
			codec.Reset()
			continue
		}
		codec.Seq.Update(seq)
		if len(data) == 8 {
			index := binary.LittleEndian.Uint64(data)
			c.lastIndex.Set(index)
		}
		codec.Reset()
	}
}

// Send 发送数据, 提供异步等待接受数据
func (c *Conn) Send(data []byte) *RemoteFuture {
	for !c.start {
		time.Sleep(time.Millisecond)
	}
	u := atomic.AddUint32(&c.round, 1) % c.maxConn
	encode, pos, err := c.codecs[u].Encode(data)
	if err != nil {
		return nil
	}
	conn := c.connections[u]
	if conn == nil {
		return nil
	}

	snap := encode[0:]
	size := len(encode)
	for size > 0 {
		m, err2 := conn.Write(encode)
		if err2 != nil {
			encode = snap[0:]
			size = 0
			return nil
		}
		size -= m
		encode = encode[m:]
	}
	return &RemoteFuture{seq: c.codecs[u].Seq, pos: pos, conn: conn}
}

// Call 发送并接受
func (c *Conn) Call(data []byte) ([]byte, error) {
	future := c.Send(data)
	if c.lastIndex == nil {
		// 同步
		codec := NewHeadBodyCodec()
		res, _, err := codec.Decode(future.conn)
		for err != nil {
			if err == ErrIncompletePacket {
				res, _, err = codec.Decode(future.conn)
				continue
			}
			break
		}
		return res, err
	} else {
		future.WaitDone()
		return nil, nil
	}
}

func (c *Conn) Reconnect(u uint32) net.Conn {
	conn := c.connections[u]
	if conn != nil {
		_ = conn.Close()
	}
	c.connections[u] = nil
	newConn, err := net.DialTimeout("tcp", c.target, ConnectTimeout)
	for err != nil {
		time.Sleep(ConnectTimeout)
		newConn, err = net.DialTimeout("tcp", c.target, ConnectTimeout)
	}
	logging.Infof("[%d]connecting %s success", u, c.target)
	c.connections[u] = newConn
	return newConn
}

func (c *Conn) Close() error {
	for i := 0; i < int(c.maxConn); i++ {
		if c.connections[i] != nil {
			_ = c.connections[i].Close()
		}
	}
	return nil
}

// TcpClient
// 检查服务端健康状态
// 断开重连
// 池化tcp连接
type TcpClient struct {
	sync.Mutex
	connMap        map[string]*Conn
	maxConnPerHost int
	lastIndex      *common.LastIndex
}

func NewTcpClient(urls []string, maxConnPerHost int, lastIndex *common.LastIndex) *TcpClient {
	client := &TcpClient{
		connMap:        make(map[string]*Conn),
		maxConnPerHost: maxConnPerHost,
		lastIndex:      lastIndex,
	}
	for i := 0; i < len(urls); i++ {
		go client.GetConn(urls[i])
	}
	return client
}

func (c *TcpClient) GetConn(target string) *Conn {
	conn, ok := c.connMap[target]
	if ok {
		return conn
	}
	c.Lock()
	conn, ok = c.connMap[target]
	if ok {
		c.Unlock()
		return conn
	}
	conn = NewConn(target, c.maxConnPerHost, c.lastIndex)
	c.connMap[target] = conn
	c.Unlock()
	return conn
}
